package com.cskaoyan.mall.order.mq;

import com.alibaba.fastjson2.JSON;
import com.cskaoyan.mall.common.constant.RedisConst;
import com.cskaoyan.mall.mq.constant.MqTopicConst;
import com.cskaoyan.mall.order.query.OrderInfoParam;
import com.cskaoyan.mall.order.service.OrderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

@Component
@Slf4j
public class PromoOrderConsumer {
    @Value("${rocketmq.namesrv.addr}")
    String namesrvAddr;

    @Value("${rocketmq.producer.group}")
    String producerGroup;

    @Autowired
    RedissonClient redissonClient;

    @Autowired
    OrderService orderService;

    @PostConstruct
    public void init() throws MQClientException {
        //创建消费者，消费对应的商品，然后在redis内打标记
        //咋写啊，，记不住一点，，，
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("promo-order-consumer-group");

        //设置命名中心
        consumer.setNamesrvAddr(namesrvAddr);

        //订阅主体
        consumer.subscribe(MqTopicConst.SECKILL_GOODS_QUEUE_TOPIC,"*");

        //设置回调内容
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println("运行了消费者的回调");
                //要设置成功和失败两种状态，还要防止多次回调,这里可以用的是stack和hash两种数据结构
                MessageExt messageExt = list.get(0);
                byte[] body = messageExt.getBody();
                String orderInfoStr = new String(body);           // 获取消息内容
                OrderInfoParam orderInfoParam = JSON.parseObject(orderInfoStr, OrderInfoParam.class);

                String msgId = list.get(0).getMsgId();

                System.out.println("运行了消费者的回调"+msgId+"skuId"+orderInfoParam.getOrderDetailList().get(0).getSkuId());

                RMap<String, String> map = redissonClient.getMap(RedisConst.PROMO_CONSUMER_STATE);
                boolean existRun = map.fastPut(msgId, msgId);
                if (!existRun){
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                //然后调用对应service方法呗
                try {
                    Long skuId = orderService.saveSeckillOrder(orderInfoParam);
                    if (skuId!=null){
                        //这个地方还要存入redis里面orderId，值是skuId+userId
                        RMap<String, Long> submitMap = redissonClient.getMap(RedisConst.PROMO_SUBMIT_ORDER);

                        String key = orderInfoParam.getUserId() + "" + orderInfoParam.getOrderDetailList().get(0).getSkuId();
                        submitMap.put(key, skuId);
                        System.out.println("消费者回调成功");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                }catch (Exception e){
                    e.printStackTrace();
                }
                map.remove(msgId);
                System.out.println("消费者回调失败");
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;

            }
        });

        consumer.start();
    }
}
